package tableapi

import bean.SensorReading
import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.{StreamTableEnvironment, _}
import org.apache.flink.table.api.{DataTypes, Table}
import org.apache.flink.table.descriptors.{Csv, Elasticsearch, FileSystem, Json, Kafka, Schema}

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/11/27 12:14
  */
object Table6_EsOutput {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    //file source
    val inputPath: String = "src/main/resources/sensor.txt"
    val fileStream: DataStream[String] = env.readTextFile(inputPath)

    val socketStream = env.socketTextStream("192.168.88.106", 7777)

    //先转换为特定的类型
    val dataStream: DataStream[SensorReading] = fileStream.map(data => {
      val arr = data.split(",")
      SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
    })

    //创建表执行环境
    val tableEnv = StreamTableEnvironment.create(env)
    //连接外部File系统，读取数据，注册表
    tableEnv.connect(new FileSystem()
      .path(inputPath)
    )
      //      .withFormat(new OldCsv())
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temp", DataTypes.DOUBLE())
      )
      .createTemporaryTable("fileTable")

    //转换操作
    val sensorTable: Table = tableEnv.from("fileTable")
    val resultTable: Table = sensorTable
      .select('id, 'temp)
      .filter('id === "sensor_1")
    //聚合转换
    val aggTable: Table = sensorTable
      .groupBy('id)
      .select('id, 'id.count as 'count)
    resultTable.toAppendStream[(String,Double)].print("result")
    aggTable.toRetractStream[(String,Long)].print("agg")

    //输出到ES
    tableEnv.connect(new Elasticsearch()
      .version("6")
        .host("192.168.88.106",9200,"http")
        .index("sensor1")
        .documentType("temp")
    )
      .inUpsertMode()
      .withFormat(new Json)
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("count", DataTypes.BIGINT())
      )
      .createTemporaryTable("EsOutputTable")

    aggTable.insertInto("EsOutputTable")



    env.execute("table api test")
  }


}
